package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {

	start := time.Now()
	done := make(chan struct{})
	defer close(done)

	results := make([]<-chan int, 0)

	for data := range gener(done, 10) {

		go func(data int) {
			ch := make(chan int)
			defer close(ch)
			results = append(results, doSome(done, ch))
			ch <- data
		}(data)
	}

	//fmt.Println(len(results))

	for i := range fanIn(done, results...) {
		fmt.Println(i)
	}

	fmt.Println(time.Now().Sub(start))

}

func gener(done <-chan struct{}, n int) <-chan int {
	result := make(chan int)

	go func() {
		defer close(result)
		for i := 0; i < n; i++ {
			select {
			case result <- 1 + i:
			case <-done:
				return
			}
		}
	}()

	return result
}

func doSome(done <-chan struct{}, stream <-chan int) <-chan int {
	result := make(chan int)

	go func() {
		defer close(result)
		for i := range stream {
			time.Sleep(time.Second * 2) // 耗时的计算
			select {
			case result <- 2 * i:
			case <-done:
				return
			}
		}
	}()
	return result
}

func fanIn(done <-chan struct{}, streams ...<-chan int) <-chan int {

	result := make(chan int)
	g := sync.WaitGroup{}

	for i := range streams {
		g.Add(1)
		go func(i int) {
			defer g.Done()
			select {
			case result <- <-streams[i]:
			case <-done:
				return
			}
		}(i)
	}

	go func() {
		g.Wait()
		close(result)
	}()

	return result
}
